/**
 * Copyright (C) 2010-2013 Alibaba Group Holding Limited
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.alibaba.rocketmq.store.schedule;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.rocketmq.common.ConfigManager;
import com.alibaba.rocketmq.common.TopicFilterType;
import com.alibaba.rocketmq.common.constant.LoggerName;
import com.alibaba.rocketmq.common.message.MessageAccessor;
import com.alibaba.rocketmq.common.message.MessageConst;
import com.alibaba.rocketmq.common.message.MessageDecoder;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.running.RunningStats;
import com.alibaba.rocketmq.store.ConsumeQueue;
import com.alibaba.rocketmq.store.DefaultMessageStore;
import com.alibaba.rocketmq.store.MessageExtBrokerInner;
import com.alibaba.rocketmq.store.PutMessageResult;
import com.alibaba.rocketmq.store.PutMessageStatus;
import com.alibaba.rocketmq.store.SelectMapedBufferResult;
import com.alibaba.rocketmq.store.config.StorePathConfigHelper;

/**
 * 定时消息服务
 * =======================1、解析并存储delayOffset.json文件========================
 * =======================2、定时执行延迟消息的任务。===============================
 * @author shijia.wxr<vintage.wang@gmail.com>
 * @since 2013-7-21
 */
public class ScheduleMessageService extends ConfigManager {
	public static final String SCHEDULE_TOPIC = "SCHEDULE_TOPIC_XXXX";
	private static final Logger log = LoggerFactory.getLogger(LoggerName.StoreLoggerName);
	private static final long FIRST_DELAY_TIME = 1000L;
	private static final long DELAY_FOR_A_WHILE = 100L;
	private static final long DELAY_FOR_A_PERIOD = 10000L;
	// 每个level对应的延时时间
	private final ConcurrentHashMap<Integer /* level */, Long/* delay timeMillis */> delayLevelTable = new ConcurrentHashMap<Integer, Long>(32);
	// 延时计算到了哪里
	private final ConcurrentHashMap<Integer /* level */, Long/* offset */> offsetTable = new ConcurrentHashMap<Integer, Long>(32);
	// 定时器
	private final Timer timer = new Timer("ScheduleMessageTimerThread", true);
	// 存储顶层对象
	private final DefaultMessageStore defaultMessageStore;
	// 最大值
	private int maxDelayLevel;

	public ScheduleMessageService(final DefaultMessageStore defaultMessageStore) {
		this.defaultMessageStore = defaultMessageStore;
	}

	public void buildRunningStats(HashMap<String, String> stats) {
		Iterator<Entry<Integer, Long>> it = this.offsetTable.entrySet().iterator();
		while (it.hasNext()) {
			Entry<Integer, Long> next = it.next();
			int queueId = delayLevel2QueueId(next.getKey());
			long delayOffset = next.getValue();
			long maxOffset = this.defaultMessageStore.getMaxOffsetInQuque(SCHEDULE_TOPIC, queueId);
			String value = String.format("%d,%d", delayOffset, maxOffset);
			String key = String.format("%s_%d", RunningStats.scheduleMessageOffset.name(), next.getKey());
			stats.put(key, value);
		}
	}

	public static int queueId2DelayLevel(final int queueId) {
		return queueId + 1;
	}

	public static int delayLevel2QueueId(final int delayLevel) {
		return delayLevel - 1;
	}

	private void updateOffset(int delayLevel, long offset) {
		this.offsetTable.put(delayLevel, offset);
	}

	public long computeDeliverTimestamp(final int delayLevel, final long storeTimestamp) {
		Long time = this.delayLevelTable.get(delayLevel);
		if (time != null) {
			return time + storeTimestamp;
		}

		return storeTimestamp + 1000;
	}

	/**
	 * =================在Broker启动时调用ScheduleMessageService.start方法=======================
	 */
	public void start() {
		// 为每个延时队列增加定时器
		for (Integer level : this.delayLevelTable.keySet()) {
			Long timeDelay = this.delayLevelTable.get(level);
			// 以level从offsetTable中获取该级别的队列消费进度offset
			Long offset = this.offsetTable.get(level);
			if (null == offset) {
				offset = 0L;
			}

			if (timeDelay != null) {
				// 以level、offset为参数初始化DeliverDelayedMessageTimerTask对象，并存入定时器中，在1秒之后在调用该对象的run方法
				this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
			}
		}

		// 设置定时任务将延时进度刷盘
		this.timer.scheduleAtFixedRate(new TimerTask() {

			@Override
			public void run() {
				try {
					ScheduleMessageService.this.persist();
				} catch (Exception e) {
					log.error("scheduleAtFixedRate flush exception", e);
				}
			}
		}, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval());
	}

	public void shutdown() {
		this.timer.cancel();
	}

	public int getMaxDelayLevel() {
		return maxDelayLevel;
	}

	public String encode() {
		return this.encode(false);
	}

	public String encode(final boolean prettyFormat) {
		DelayOffsetSerializeWrapper delayOffsetSerializeWrapper = new DelayOffsetSerializeWrapper();
		delayOffsetSerializeWrapper.setOffsetTable(this.offsetTable);
		return delayOffsetSerializeWrapper.toJson(prettyFormat);
	}

	@Override
	public void decode(String jsonString) {
		if (jsonString != null) {
			DelayOffsetSerializeWrapper delayOffsetSerializeWrapper = DelayOffsetSerializeWrapper.fromJson(jsonString, DelayOffsetSerializeWrapper.class);
			if (delayOffsetSerializeWrapper != null) {
				this.offsetTable.putAll(delayOffsetSerializeWrapper.getOffsetTable());
			}
		}
	}

	@Override
	public String configFilePath() {
		return StorePathConfigHelper.getDelayOffsetStorePath(this.defaultMessageStore.getMessageStoreConfig().getStorePathRootDir());
	}

	public boolean load() {
		// 加载delayOffset.json文件内容
		boolean result = super.load();
		// 将这些级别（"1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"）的延时
		// 存入延迟级别delayLevelTable：ConcurrentHashMap<Integer /* level */, Long/* delay timeMillis */>变量中，
		// 例如1s的kv值为1:1000,5s的kv值为2:5000，key值依次类推；每个延迟级别即为一个队列。
		result = result && this.parseDelayLevel();
		return result;
	}

	public boolean parseDelayLevel() {
		HashMap<String, Long> timeUnitTable = new HashMap<String, Long>();
		timeUnitTable.put("s", 1000L);// 秒
		timeUnitTable.put("m", 1000L * 60);// 分钟
		timeUnitTable.put("h", 1000L * 60 * 60);// 小时
		timeUnitTable.put("d", 1000L * 60 * 60 * 24);// 天

		// "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h"
		String levelString = this.defaultMessageStore.getMessageStoreConfig().getMessageDelayLevel();
		try {
			String[] levelArray = levelString.split(" ");
			for (int i = 0; i < levelArray.length; i++) {
				String value = levelArray[i];
				String ch = value.substring(value.length() - 1);
				Long tu = timeUnitTable.get(ch);

				int level = i + 1;
				if (level > this.maxDelayLevel) {
					this.maxDelayLevel = level;
				}
				long num = Long.parseLong(value.substring(0, value.length() - 1));
				long delayTimeMillis = tu * num;
				this.delayLevelTable.put(level, delayTimeMillis);
			}
		} catch (Exception e) {
			log.error("parseDelayLevel exception", e);
			log.info("levelString String = {}", levelString);
			return false;
		}

		return true;
	}

	class DeliverDelayedMessageTimerTask extends TimerTask {
		private final int delayLevel;
		private final long offset;

		public DeliverDelayedMessageTimerTask(int delayLevel, long offset) {
			this.delayLevel = delayLevel;
			this.offset = offset;
		}

		@Override
		public void run() {
			try {
				// 延迟消费逻辑
				this.executeOnTimeup();
			} catch (Exception e) {
				// XXX: warn and notify me
				log.error("ScheduleMessageService, executeOnTimeup exception", e);
				// 若该方法抛出异常，则重新初始化DeliverDelayedMessageTimerTask对象并放入定时器中，在10秒之后重新执行该对象
				ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel, this.offset), DELAY_FOR_A_PERIOD);
			}
		}

		/**
		 * 纠正下次投递时间，如果时间特别大，则纠正为当前时间
		 * 
		 * @return
		 */
		private long correctDeliverTimestamp(final long now, final long deliverTimestamp) {
			// 如果为0，则会立刻投递
			long result = deliverTimestamp;
			// 超过最大值，纠正为当前时间
			long maxTimestamp = now + ScheduleMessageService.this.delayLevelTable.get(this.delayLevel);
			if (deliverTimestamp > maxTimestamp) {
				result = now;
			}

			return result;
		}

		public void executeOnTimeup() {
			// 调用DefaultMessageStore对象的findConsumeQueue(String topic, int queueId)方法获取对应的ConsumeQueue对象；
			// 其中topic等于"SCHEDULE_TOPIC_XXXX"；queueId等于延迟级别level减1；
			ConsumeQueue cq = ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(SCHEDULE_TOPIC, delayLevel2QueueId(delayLevel));

			long failScheduleOffset = offset;

			if (cq != null) {
				// 以offset偏移量为参数调用ConsumeQueue对象的getIndexBuffer方法获取consumequeue的数据
				SelectMapedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
				if (bufferCQ != null) {
					try {
						long nextOffset = offset;
						int i = 0;
						// 从consumequeue的数据中逐个遍历每个消息单元(每个消息单元为20字节)
						for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQStoreUnitSize) {
							long offsetPy = bufferCQ.getByteBuffer().getLong(); // commitLog Offset(8)
							int sizePy = bufferCQ.getByteBuffer().getInt(); // size(4)
							long tagsCode = bufferCQ.getByteBuffer().getLong(); // Message Tag HashCode(8)

							// 队列里存储的tagsCode实际是一个时间点
							long now = System.currentTimeMillis();
							long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);

							nextOffset = offset + (i / ConsumeQueue.CQStoreUnitSize);

							long countdown = deliverTimestamp - now;
							// 时间到了，该投递
							if (countdown <= 0) {
								MessageExt msgExt = ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(offsetPy, sizePy);

								if (msgExt != null) {
									try {
										// 将真实的topic和queueId从消息的properties中取出并设置到给MessageExtBrokerInner对象中
										MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
										// 消息写入
										PutMessageResult putMessageResult = ScheduleMessageService.this.defaultMessageStore.putMessage(msgInner);
										// 成功
										if (putMessageResult != null && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
											continue;
										}
										// 失败
										else {
											// XXX: warn and notify me
											log.error("ScheduleMessageService, a message time up, but reput it failed, topic: {} msgId {}", msgExt.getTopic(), msgExt.getMsgId());
											// 以该数据开始偏移量为offset、对应的level为参数初始化DeliverDelayedMessageTimerTask对象并放入定时器中，在10秒之后重新执行
											ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset), DELAY_FOR_A_PERIOD);
											// 更新ScheduleMessageService.offsetTable变量中相应延迟级别的offset值
											ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
											return;
										}
									} catch (Exception e) {
										/*
										 * XXX: warn and notify me msgExt里面的内容不完整 ，如没有REAL_QID,REAL_TOPIC之类的 ，导致数据无法正常的投递到正确的消费队列，所以暂时先直接跳过该条消息
										 */
										log.error("ScheduleMessageService, messageTimeup execute error, drop it. msgExt=" + msgExt + ", nextOffset=" + nextOffset + ",offsetPy=" + offsetPy + ",sizePy=" + sizePy, e);
									}
								}
							}
							// 时候未到，继续定时
							else {
								ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset), countdown);
								ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
								return;
							}
						} // end of for

						// 将上述consumequeue数据的最后一个消息读取偏移量之后的一个消息读取偏移量（offset+consumequeue消息中的消息单元个数）作为nextoffset
						nextOffset = offset + (i / ConsumeQueue.CQStoreUnitSize);
						// 对应的level为参数初始化DeliverDelayedMessageTimerTask对象并放入定时器中，100毫秒之后再重新执行；
						ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset), DELAY_FOR_A_WHILE);
						ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
						return;
					} finally {
						// 必须释放资源
						bufferCQ.release();
					}
				} // end of if (bufferCQ != null)
				else {
					/*
					 * 索引文件被删除，定时任务中记录的offset已经被删除，会导致从该位置中取不到数据， 这里直接纠正下一次定时任务的offset为当前定时任务队列的最小值
					 */
					long cqMinOffset = cq.getMinOffsetInQuque();
					if (offset < cqMinOffset) {
						failScheduleOffset = cqMinOffset;
						log.error("schedule CQ offset invalid. offset=" + offset + ", cqMinOffset=" + cqMinOffset + ", queueId=" + cq.getQueueId());
					}
				}
			} // end of if (cq != null)

			// 若未获取到ConsumeQueue对象则以offset、level为参数初始化DeliverDelayedMessageTimerTask对象并放入定时器中，100毫秒之后再重新执行；
			ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel, failScheduleOffset), DELAY_FOR_A_WHILE);
		}

		private MessageExtBrokerInner messageTimeup(MessageExt msgExt) {
			MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
			msgInner.setBody(msgExt.getBody());
			msgInner.setFlag(msgExt.getFlag());
			MessageAccessor.setProperties(msgInner, msgExt.getProperties());

			TopicFilterType topicFilterType = MessageExt.parseTopicFilterType(msgInner.getSysFlag());
			long tagsCodeValue = MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());
			msgInner.setTagsCode(tagsCodeValue);
			msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));

			msgInner.setSysFlag(msgExt.getSysFlag());
			msgInner.setBornTimestamp(msgExt.getBornTimestamp());
			msgInner.setBornHost(msgExt.getBornHost());
			msgInner.setStoreHost(msgExt.getStoreHost());
			msgInner.setReconsumeTimes(msgExt.getReconsumeTimes());

			msgInner.setWaitStoreMsgOK(false);
			MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_DELAY_TIME_LEVEL);

			// 恢复Topic
			msgInner.setTopic(msgInner.getProperty(MessageConst.PROPERTY_REAL_TOPIC));

			// 恢复QueueId
			String queueIdStr = msgInner.getProperty(MessageConst.PROPERTY_REAL_QUEUE_ID);
			int queueId = Integer.parseInt(queueIdStr);
			msgInner.setQueueId(queueId);

			return msgInner;
		}
	}
}
